package cn.doitedu.kafka.day02

import java.io.{File, RandomAccessFile}
import java.nio.charset.Charset

import org.apache.commons.io.FileUtils

import scala.io.Source

/**
 * 自己编写一个实时的日志采集工具，并且保证数据不重复读取数据
 *
 * 实时：定期读取，比如0.5秒检测一次是否有新的数据写入
 * 不重复读取：要记录偏移量
 * 写入Kafka，保证数据的安全：开启acks = 1
 *
 */
object MyRealTimeLogCollector {

  def main(args: Array[String]): Unit = {
    var offset: Long = 0L
    //必须传入两个参数
    //读取日志的路径的名称
    val logFilePath = args(0)
    //保存日志文件偏移量的文件
    val offsetFilePath = args(1)

    //1.获取偏移量文件中的历史偏移量
    val offsetFile = new File(offsetFilePath)
    //如果保存偏移量的文件不存在，先创建
    if(!offsetFile.exists()) {
      offsetFile.createNewFile()
    }
    //2.如果历史偏移量不为空，接着以前的偏移量继续读取，如果为空就从0开始读取
    val offsetStr = FileUtils.readFileToString(offsetFile, Charset.forName("UTF-8"))
    if (offsetStr != null && !"".equals(offsetStr)) {
      //以前保存过偏移量，把以前的偏移量读取出来，赋值给offset
      offset = offsetStr.toLong
    }

    //3.创建一个特殊的文件RandomAccessFile，可以从指定的偏移量继续读取数据
    val raf = new RandomAccessFile(logFilePath, "r") //r代表读取 rw代表读写
    //RandomAccessFile的seek方法可以从指定偏移量开始读取数据
    raf.seek(offset)

    //4.读出来一行，然后将数据写入到Kafka
    //以后改进（读取10行数据，批量写入到Kafka）
    while (true) {

      var line = raf.readLine()
      //有新的内容写入
      if(line != null) {
        line = new String(line.getBytes(Charset.forName("ISO-8859-1")), "UTF-8")
        //调用Kafka的API将数据写入到Kafka中
        KafkaUtils.writeDataToKafka("test2", line)
        //获取最新的偏移量
        offset = raf.getFilePointer
        //更新偏移量
        FileUtils.writeStringToFile(offsetFile, offset.toString, Charset.forName("UTF-8"))
      } else {
        //没有新的数据写入
        Thread.sleep(500)
      }

    }





    //然后继续读取数据










  }

}
